package net.bwie.realtime.vehicle.dwd.function;

import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.vehicle.dwd.bean.MonitorInfo;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author ASUS
 */
public class LateDataProcessFunction extends ProcessFunction<MonitorInfo, String> {
    @Override
    public void processElement(MonitorInfo monitorInfo, ProcessFunction<MonitorInfo, String>.Context context, Collector<String> collector) throws Exception {
        // 构建迟到数据信息
        JSONObject lateData = new JSONObject();
        lateData.put("car", monitorInfo.getCar());
        lateData.put("speed", monitorInfo.getSpeed());
        lateData.put("speedLimit", monitorInfo.getSpeedLimit());
        lateData.put("actionTime", monitorInfo.getActionTime());
        lateData.put("roadId", monitorInfo.getRoadId());
        lateData.put("monitorId", monitorInfo.getMonitorId());
        lateData.put("processTime", System.currentTimeMillis());
        // 标记为迟到数据流
        lateData.put("processType", "late_stream");
        lateData.put("lateReason", "数据迟到，超过窗口关闭时间");

        collector.collect(lateData.toJSONString());
    }
}
